package com.gugee.tiktok.data.consumer.blogger.listener;

import com.gugee.tiktok.data.common.msg.BloggerMessage;
import com.gugee.tiktok.data.consumer.blogger.core.main.SaveFansPortrait;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class BloggerMessageForFansPortraitListener {

    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    SaveFansPortrait saveFansPortrait;

    @KafkaListener(id="${kafka.consumer.group.blogger.fansportrait.id}",topics = {"${kafka.topics.blogger.fansportrait.online}"},containerFactory = "factoryBloggerAck")
    public void listen(List<BloggerMessage> messages, Acknowledgment ack) {
        long start=System.currentTimeMillis();
        String error="success";
        try {
            logger.info("接收到消息数量: {}" , messages.size());
            saveFansPortrait.handle(messages);
            ack.acknowledge();
        }catch (Exception e){
//            e.printStackTrace();
            logger.info("发生异常: {}" , e.getMessage());
            error=e.getMessage();
        }finally {
            long times=(System.currentTimeMillis()-start)/1000;
            if(times<1){
                times=1;
            }
            logger.info("消息数量: {},处理耗时: {}s,每秒处理消息数={}/s,结果：{}" , messages.size(),times,messages.size()/times,error);
        }
    }
}
